package org.example.reader;

import org.apache.paimon.Snapshot;
import org.apache.paimon.format.FileFormat;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.manifest.ManifestList;
import org.apache.paimon.stats.SimpleStats;
import org.apache.paimon.utils.SnapshotManager;
import org.apache.paimon.format.avro.AvroFileFormat;
import org.apache.paimon.types.RowType;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.utils.FileStorePathFactory;
import org.apache.paimon.format.FileFormatFactory;
import org.apache.paimon.options.Options;

import java.io.IOException;
import java.util.List;

/**
 * 工具类，用于读取并打印Paimon中的manifest list文件内容
 */
public class ManifestListReader {

    /**
     * 使用Paimon API读取指定表和snapshot的manifest list
     *
     * @param tablePath Paimon表的根路径
     * @param snapshotId 要读取的snapshot ID
     */
    public static void readManifestListWithPaimonAPI(String tablePath, long snapshotId) throws IOException {
        System.out.println("使用Paimon API读取表路径: " + tablePath + " 的snapshot ID: " + snapshotId);

        // 创建FileIO
        FileIO fileIO = new LocalFileIO();

        // 创建Path对象
        Path path = new Path(tablePath);

        // 创建SnapshotManager
        SnapshotManager snapshotManager = new SnapshotManager(fileIO, path);

        // 获取指定的snapshot
        Snapshot snapshot = null;
        try {
            snapshot = snapshotManager.snapshot(snapshotId);
        } catch (Exception e) {
            System.err.println("无法读取snapshot " + snapshotId + ": " + e.getMessage());
            return;
        }

        // 获取baseManifestList和deltaManifestList路径
        String baseManifestListName = snapshot.baseManifestList();
        String deltaManifestListName = snapshot.deltaManifestList();
        String changelogManifestListName = snapshot.changelogManifestList();

        System.out.println("\nSnapshot信息:");
        System.out.println("ID: " + snapshot.id());
        System.out.println("Schema ID: " + snapshot.schemaId());
        System.out.println("Base Manifest List: " + baseManifestListName);
        System.out.println("Delta Manifest List: " + deltaManifestListName);
        System.out.println("Changelog Manifest List: " +
                (changelogManifestListName != null ? changelogManifestListName : "无"));

        // 创建一个简单的RowType（只用于构造FileStorePathFactory）
        RowType rowType = RowType.of(DataTypes.STRING());

        // 基于目录结构创建正确的FileStorePathFactory
        FileStorePathFactory pathFactory = new FileStorePathFactory(
                path,                // 表路径
                rowType,             // 表的行类型
                "default",           // 默认分区值
                "avro",              // 格式标识符
                "data",              // 数据文件前缀
                "changelog",         // 变更日志文件前缀
                false,               // 不使用旧式分区名称
                true,                // 文件后缀包含压缩信息
                "lz4",               // 压缩格式
                null                 // 数据文件路径目录
        );

        // 创建Options对象
        Options options = new Options();

        // 创建FormatContext
        FileFormatFactory.FormatContext formatContext = new FileFormatFactory.FormatContext(
                options,      // 配置选项
                1024,         // SchemaManager
                1024          // 类加载器
        );

        // 创建FileFormat - 使用AvroFileFormat类并传递FormatContext
        FileFormat fileFormat = new AvroFileFormat(formatContext);

        // 创建ManifestList
        ManifestList manifestList = new ManifestList.Factory(
                fileIO,
                fileFormat,
                "lz4", // 默认压缩格式
                pathFactory,
                null).create();

        // 使用ManifestList的readAllManifests方法直接读取所有清单文件
        System.out.println("\n所有Manifest文件信息:");
        List<ManifestFileMeta> allManifests = manifestList.readAllManifests(snapshot);
        printManifestFileMetas(allManifests);

        // 读取baseManifestList文件
        if (baseManifestListName != null && !baseManifestListName.isEmpty()) {
            System.out.println("\nBase Manifest List内容:");
            List<ManifestFileMeta> baseManifests = manifestList.read(baseManifestListName);
            printManifestFileMetas(baseManifests);
        }

        // 读取deltaManifestList文件
        if (deltaManifestListName != null && !deltaManifestListName.isEmpty()) {
            System.out.println("\nDelta Manifest List内容:");
            List<ManifestFileMeta> deltaManifests = manifestList.read(deltaManifestListName);
            printManifestFileMetas(deltaManifests);
        }

        // 读取changelogManifestList文件（如果有）
        if (changelogManifestListName != null && !changelogManifestListName.isEmpty()) {
            System.out.println("\nChangelog Manifest List内容:");
            List<ManifestFileMeta> changelogManifests = manifestList.read(changelogManifestListName);
            printManifestFileMetas(changelogManifests);
        }
    }

    /**
     * 打印ManifestFileMeta列表
     */
    private static void printManifestFileMetas(List<ManifestFileMeta> manifests) {
        if (manifests == null || manifests.isEmpty()) {
            System.out.println("没有发现manifest记录");
            return;
        }

        int count = 0;
        for (ManifestFileMeta meta : manifests) {
            count++;
            System.out.println("\n记录 #" + count);
            System.out.println("--------------------------------------------------------");
            System.out.println("文件名: " + meta.fileName());
            System.out.println("文件大小: " + meta.fileSize() + " 字节");
            System.out.println("新增文件数: " + meta.numAddedFiles());
            System.out.println("删除文件数: " + meta.numDeletedFiles());
            System.out.println("Schema ID: " + meta.schemaId());

            // 打印分区统计信息
            SimpleStats stats = meta.partitionStats();
            if (stats != null) {
                System.out.println("分区统计信息:");
                System.out.println("  最小值: " + (stats.minValues() != null ? "有" : "无"));
                System.out.println("  最大值: " + (stats.maxValues() != null ? "有" : "无"));
                System.out.println("  空值计数: " + (stats.nullCounts() != null ? "有" : "无"));
            }
        }

        System.out.println("\n--------------------------------------------------------");
        System.out.println("共读取 " + count + " 条manifest记录");
    }

    /**
     * 主方法，演示如何使用这个工具
     * 参数可以是: <表路径> <snapshot ID>
     */
    public static void main(String[] args) throws IOException {
        String tablePath = "/tmp/paimon/default.db/t_user";
        long snapshotId = 1;

        // 如果有命令行参数，则使用命令行参数
        if (args.length >= 1) {
            tablePath = args[0];
        }
        if (args.length >= 2) {
            try {
                snapshotId = Long.parseLong(args[1]);
            } catch (NumberFormatException e) {
                System.err.println("无效的snapshot ID参数，使用默认值1");
            }
        }

        readManifestListWithPaimonAPI(tablePath, snapshotId);
    }
}